package cn.item.buckle.business;

import cn.item.buckle.business.pojo.*;
import cn.item.buckle.business.service.MonitorCameraStateService;
import cn.item.buckle.confg.ConfigEnum;
import cn.lsh.kafka.db.service.ServiceFactory;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.*;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.StreamSupport;

public class BuckleMonitor {
	public static final Logger log = LoggerFactory.getLogger(BuckleMonitor.class);

	public static void main(String[] args) {

		SparkSession ss = SparkSession.builder()
				.master("local")
				.enableHiveSupport()
				.config(ConfigEnum.HIVE_METASTORE_ADDR.getName(), ConfigEnum.HIVE_METASTORE_ADDR.getValue())
				.appName("buckle_monitor").getOrCreate();
		MonitorAccumulator monitorAccumulator = new MonitorAccumulator();
		//注册自定义累加器
		ss.sparkContext().register(monitorAccumulator, "MonitorAccumulator");

		ss.sql("use " + ConfigEnum.HIVE_DATABASE.getValue());
		Dataset<Row> flowData = ss.table(ConfigEnum.MONITOR_FLOW_ACTION.getValue());
		JavaPairRDD<String, Row> monitorDetailRdd = flowData.toJavaRDD().mapToPair(row -> new Tuple2<>(row.getString(1), row));
		monitorDetailRdd.cache();
		//按照卡扣分组，九个卡扣一共九组
		JavaPairRDD<String, Iterable<Row>> monitorByIdRDD = monitorDetailRdd.groupByKey();
		//按照key分组聚合,聚合后结果为：("0005","monitorId=0005|camearIds=09200,03243|cameraCount=2|carCount==100")
		// JavaPairRDD<String, String> aggregateMonitorByIdRDD = aggregateByMonitor(monitorByIdRDD);
		JavaPairRDD<String, MonitorCameraInfo> aggregateMonitorByIdRDD = aggregateByMonitor(monitorByIdRDD);
		/*@step1 监测卡扣状态*/
		JavaPairRDD<Integer, String> carCountMonitorRdd = checkMonitorState(ss, monitorAccumulator, aggregateMonitorByIdRDD);
		//action算子触发
		// carCountMonitorRdd.count();
		log.info("-------save monitor state---------");
		// saveMonitorState(ConfigEnum.APPLICATION_TASK_ID.getLongValue(), monitorAccumulator);
		/*@step2 求通过车辆数top5的卡扣*/
		// analysisMonitorCarNumTopN(ss, monitorDetailRdd);
		/*@step4 获取高速通过的车辆的topN卡扣*/
		// analysisMonitorCarSpeedTopN(monitorDetailRdd);
		/*@step5 通过指定卡扣的所有车辆的轨迹*/
		// analysisMonitorCarTrack(monitorDetailRdd);
		/*@step6 区域碰撞分析出现在不同区域的同一车辆*/
		// analysisMonitorCarArea(monitorDetailRdd);
		/*@step6 随机抽取车辆*/
		// analysisRandomTaskSample(ss, monitorDetailRdd);
		/*@step7 卡扣车流量转换率*/
		analysisMonitorTrafficPercentRate(ss, monitorDetailRdd);
		ss.close();
		log.info("-------all finished---------");
	}

	public static JavaPairRDD<String, MonitorCameraInfo> aggregateByMonitor(JavaPairRDD<String, Iterable<Row>> monitorByIdRDD) {
		return monitorByIdRDD.aggregateByKey(new MonitorCameraInfo(), new Function2<MonitorCameraInfo, Iterable<Row>, MonitorCameraInfo>() {
			private static final long serialVersionUID = 6729320784553395926L;

			@Override
			public MonitorCameraInfo call(MonitorCameraInfo curr, Iterable<Row> next) throws Exception {
				String key = null;
				Set<String> camearIds = new HashSet<>();
				int carCount = 0;
				for (Row row : next) {
					carCount++;
					if (key == null) {
						key = row.getString(1);
					}
					camearIds.add(row.getString(2));
				}
				if (StringUtils.isBlank(curr.getMonitorId())) {
					curr.setMonitorId(key);
				}
				curr.addCamearIds(camearIds);
				curr.addCameraCount(camearIds.size());
				curr.addCarCount(carCount);
				return curr;
			}
		}, new Function2<MonitorCameraInfo, MonitorCameraInfo, MonitorCameraInfo>() {
			private static final long serialVersionUID = 6210152697577338922L;

			@Override
			public MonitorCameraInfo call(MonitorCameraInfo last, MonitorCameraInfo next) throws Exception {
				last.addCamearIds(next.getCamearIds());
				last.addCameraCount(next.getCameraCount());
				last.addCarCount(next.getCarCount());
				return last;
			}
		});
	}

	public static JavaPairRDD<String, String> aggregateByMonitorOld(JavaPairRDD<String, Iterable<Row>> monitorByIdRDD) {
		return monitorByIdRDD.aggregateByKey("", new Function2<String, Iterable<Row>, String>() {
			private static final long serialVersionUID = -1346837355755634115L;

			@Override
			public String call(String last, Iterable<Row> values) throws Exception {
				String ret = "monitorId=";
				String key = null;
				Set<String> camearIds = new HashSet<>();
				int carCount = 0;
				for (Row row : values) {
					carCount++;
					if (key == null) {
						key = row.getString(1);
					}
					camearIds.add(row.getString(2));
				}
				return ret + key + "|" + "camearIds=" + String.join(",", camearIds) + "|cameraCount=" + camearIds.size() + "|carCount=" + carCount;
			}
		}, new Function2<String, String, String>() {
			private static final long serialVersionUID = -6162709014690136948L;

			@Override
			public String call(String lastPartitionValue, String nextPartitionValue) throws Exception {
				String[] lastSplit = lastPartitionValue.split("\\|");
				List<String> lastCameraIds = Arrays.asList(lastSplit[1].split("=")[2].split(","));
				int lastCarCount = Integer.parseInt(lastSplit[3].split("=")[2]);
				String[] nextSplit = nextPartitionValue.split("\\|");
				int nextCarCount = Integer.parseInt(nextSplit[3].split("=")[2]);
				String[] nextCameraIds = lastSplit[1].split("=")[2].split(",");
				for (String nextCameraId : nextCameraIds) {
					if (!lastCameraIds.contains(nextCameraId)) {
						lastCameraIds.add(nextCameraId);
					}
				}
				int carCount = lastCarCount + nextCarCount;
				return lastSplit[0] + "|" + "camearIds=" + String.join(",", lastCameraIds) + "|cameraCount=" + lastCameraIds.size() + "|carCount=" + carCount;
			}
		});
	}

	/**
	 * 1、分析通过车辆数最多的top5卡扣
	 * 2、统计top5卡扣下经过的所有车辆详细信息
	 *
	 * @param ss
	 * @param monitorDetailRdd
	 */
	public static void analysisMonitorCarNumTopN(SparkSession ss, JavaPairRDD<String, Row> monitorDetailRdd) {
		List<Tuple2<String, Integer>> top5 = monitorDetailRdd.aggregateByKey(0, (v1, v2) -> ++v1, Integer::sum).takeOrdered(ConfigEnum.MONITOR_CAR_COUNT_TOP_N.getIntValue(), new TupleCompertor());
		List<MonitorCarCount> top5CarCounts = top5.stream().map(t -> new MonitorCarCount(ConfigEnum.APPLICATION_TASK_ID.getLongValue(), t._1, t._2)).collect(Collectors.toList());
		saveCarCount(top5CarCounts);

		List<String> top5MonitorId = top5.stream().map(Tuple2::_1).collect(Collectors.toList());
		//可使用广播变量避免用join而产生shuffle
		Broadcast<List<String>> top5Broadcast = new JavaSparkContext(ss.sparkContext()).broadcast(top5MonitorId);
		/*@step3 求通过车辆数top5的卡扣的车辆的详细信息*/
		monitorDetailRdd.filter(t -> top5Broadcast.getValue().contains(t._1)).foreachPartition(new VoidFunction<Iterator<Tuple2<String, Row>>>() {
			private static final long serialVersionUID = -3723704624938459537L;

			@Override
			public void call(Iterator<Tuple2<String, Row>> tuple2Iterator) throws Exception {
				List<MonitorFlowAction> result = new ArrayList<>();
				SimpleDateFormat sdf2 = new SimpleDateFormat(ConfigEnum.MONITOR_ACTION_TIME_TYPE.getValue());
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, Row> t = tuple2Iterator.next();
					MonitorFlowAction flowAction = new MonitorFlowAction();
					flowAction.setTaskId(ConfigEnum.APPLICATION_TASK_ID.getLongValue());
					flowAction.setMonitorId(t._1);
					//这里的数据类型和hive表定义的保持一致
					flowAction.setDate(t._2.getDate(0));
					flowAction.setCameraId(t._2.getString(2));
					flowAction.setCar(t._2.getString(3));
					flowAction.setActionTime(sdf2.parse(t._2.getString(4)));
					flowAction.setSpeed(t._2.getInt(5));
					flowAction.setRoadId(t._2.getString(6));
					flowAction.setAreaId(t._2.getString(7));
					result.add(flowAction);
				}
				saveMonitorState(result);
			}
		});
	}

	/**
	 * 获取高速通过的车辆的topN卡扣
	 *
	 * @param monitorDetailRdd
	 */
	public static void analysisMonitorCarSpeedTopN(JavaPairRDD<String, Row> monitorDetailRdd) {
		JavaPairRDD<String, MonitorCarSpeedCount> monitorCarSpeedRdd = monitorDetailRdd.aggregateByKey(new MonitorCarSpeedCount(), new Function2<MonitorCarSpeedCount, Row, MonitorCarSpeedCount>() {
			private static final long serialVersionUID = -609985838800028275L;

			@Override
			public MonitorCarSpeedCount call(MonitorCarSpeedCount v1, Row v2) throws Exception {
				v1.addSpeed(v2.getInt(5));
				return v1;
			}
		}, new Function2<MonitorCarSpeedCount, MonitorCarSpeedCount, MonitorCarSpeedCount>() {
			private static final long serialVersionUID = 8757610456169358240L;

			@Override
			public MonitorCarSpeedCount call(MonitorCarSpeedCount v1, MonitorCarSpeedCount v2) throws Exception {
				v1.setHigh(v1.getHigh() + v2.getHigh());
				v1.setMedium(v1.getMedium() + v2.getMedium());
				v1.setNormal(v1.getNormal() + v2.getMedium());
				v1.setLow(v1.getLow() + v2.getLow());
				return v1;
			}
		});
		List<Tuple2<MonitorCarSpeedCount, String>> carSpeedMonitorTopN = monitorCarSpeedRdd.mapToPair(t -> new Tuple2<>(t._2, t._1))
				.sortByKey(false).take(ConfigEnum.MONITOR_CAR_SPEED_TOP_N.getIntValue());
		List<MonitorCarSpeedCount> list = carSpeedMonitorTopN.stream().map(t -> {
			t._1.setTaskId(ConfigEnum.APPLICATION_TASK_ID.getLongValue());
			t._1.setMonitorId(t._2);
			return t._1;
		}).collect(Collectors.toList());
		saveList(list);
	}

	/**
	 * 分析通过指定卡口的所有车辆的轨迹
	 *
	 * @param monitorDetailRdd
	 */
	public static void analysisMonitorCarTrack(JavaPairRDD<String, Row> monitorDetailRdd) {
		JavaPairRDD<String, Tuple2<String, Date>> carMonitorDateRdd = monitorDetailRdd.mapToPair(t -> new Tuple2<>(t._2.getString(3),
				new Tuple2(t._1, new SimpleDateFormat(ConfigEnum.MONITOR_ACTION_TIME_TYPE.getValue()).parse(t._2.getString(4)))));
		carMonitorDateRdd.groupByKey().filter(new Function<Tuple2<String, Iterable<Tuple2<String, Date>>>, Boolean>() {
			private static final long serialVersionUID = 8682989715707360570L;

			@Override
			public Boolean call(Tuple2<String, Iterable<Tuple2<String, Date>>> values) throws Exception {
				boolean flag = false;
				for (Tuple2<String, Date> t : values._2) {
					if (ConfigEnum.MONITOR_CAR_TRACK_MONITOR.getValue().equals(t._1)) {
						flag = true;
						break;
					}
				}
				return flag;
			}
		}).coalesce(ConfigEnum.MONITOR_CAR_TRACK_PARTITION.getIntValue(), false)
				.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Iterable<Tuple2<String, Date>>>>>() {
					private static final long serialVersionUID = -5074153193574072249L;

					@Override
					public void call(Iterator<Tuple2<String, Iterable<Tuple2<String, Date>>>> tuple2Iterator) throws Exception {
						List<MonitorCarTrack> monitorCarTracks = new ArrayList<>();
						SimpleDateFormat sdf = new SimpleDateFormat(ConfigEnum.MONITOR_ACTION_TIME_TYPE.getValue());
						while (tuple2Iterator.hasNext()) {
							MonitorCarTrack carTrack = new MonitorCarTrack();
							Tuple2<String, Iterable<Tuple2<String, Date>>> t = tuple2Iterator.next();
							carTrack.setTaskId(ConfigEnum.APPLICATION_TASK_ID.getLongValue());
							carTrack.setMonitorId(ConfigEnum.MONITOR_CAR_TRACK_MONITOR.getValue());
							carTrack.setCar(t._1);
							//第三个参数旨在parallel时起作用，用于合并并行处理结果
							String track = StreamSupport.stream(t._2.spliterator(), false).sorted(Comparator.comparing(o -> o._2))
									.reduce("", (s, tuple2) -> {
										if (tuple2._1.equals(ConfigEnum.MONITOR_CAR_TRACK_MONITOR.getValue())) {
											return s + "," + tuple2._1 + "|" + sdf.format(tuple2._2);
										}
										if (!"".equals(s)) {
											return s + "," + tuple2._1 + "|" + sdf.format(tuple2._2);
										}
										return s;
										//第三个参数只在并行时用到
									}, (s, s2) -> null);
							carTrack.setTrack(track.substring(1));
							monitorCarTracks.add(carTrack);
						}
						saveList(monitorCarTracks);
					}
				});
	}

	/**
	 * 分析出现在不同区域的车辆
	 *
	 * @param monitorDetailRdd
	 */
	public static void analysisMonitorCarArea(JavaPairRDD<String, Row> monitorDetailRdd) {
		JavaPairRDD<String, String> carAreaRdd = monitorDetailRdd.mapToPair(t -> new Tuple2<>(t._2.getString(3), t._2.getString(7)));
		carAreaRdd.distinct().groupByKey().filter(t -> {
			Iterator<String> iterator = t._2.iterator();
			iterator.next();
			return iterator.hasNext();
		}).foreachPartition(tuple2Iterator -> {
			List<MonitorCarArea> monitorCarAreas = new ArrayList<>();
			while (tuple2Iterator.hasNext()) {
				MonitorCarArea carArea = new MonitorCarArea();
				Tuple2<String, Iterable<String>> next = tuple2Iterator.next();
				log.debug("区域碰撞车辆-" + next._1);
				carArea.setTaskId(ConfigEnum.APPLICATION_TASK_ID.getLongValue());
				carArea.setCar(next._1);
				carArea.setAreaList(String.join(",", next._2));
				monitorCarAreas.add(carArea);
			}
			saveList(monitorCarAreas);
		});
	}

	/**
	 * 每天按时间段抽取可以权威的代表当天交通情况的车辆信息
	 *
	 * @param sparkSession
	 * @param monitorDetailRdd
	 */
	public static void analysisRandomTaskSample(SparkSession sparkSession, JavaPairRDD<String, Row> monitorDetailRdd) {
		//2020-12-24 14 小时对应的carRDD
		JavaPairRDD<String, String> timeCarRdd = monitorDetailRdd.mapToPair(t -> new Tuple2<>(t._2.getString(4).substring(0, 13), t._2.getString(3)));
		timeCarRdd.cache();
		Map<String, Long> timeCarCount = timeCarRdd.countByKey();
		//计算抽取车辆的下标
		Map<String, Map<String, List<Integer>>> carOffsetList = calculateTakeCarIndex(timeCarCount);
		//广播变量
		Broadcast<Map<String, Map<String, List<Integer>>>> broadcast = new JavaSparkContext(sparkSession.sparkContext()).broadcast(carOffsetList);
		//要抽取的小时对应车辆，每天抽取100量
		JavaPairRDD<String, String> taskTimeCarRdd = timeCarRdd.groupByKey().flatMapToPair(new PairFlatMapFunction<Tuple2<String, Iterable<String>>, String, String>() {
			private static final long serialVersionUID = -404580221213949815L;

			@Override
			public Iterator<Tuple2<String, String>> call(Tuple2<String, Iterable<String>> tuple) throws Exception {
				String[] arr = tuple._1.split(" ");
				String date = arr[0];
				String hour = arr[1];
				Map<String, Map<String, List<Integer>>> takeCarIndexInfo = broadcast.value();
				List<Integer> list = takeCarIndexInfo.get(date).get(hour);
				int i = 0;
				List<Tuple2<String, String>> tuple2List = new ArrayList<>();
				for (String car : tuple._2) {
					if (list.contains(i++)) {
						tuple2List.add(new Tuple2<>(car, tuple._1));
					}
				}
				return tuple2List.iterator();
			}
		});
		JavaPairRDD<String, Row> carRowRdd = monitorDetailRdd.mapToPair(t -> new Tuple2<>(t._2.getString(3), t._2));
		JavaPairRDD<String, Tuple2<String, Row>> join = taskTimeCarRdd.distinct().join(carRowRdd);
		JavaPairRDD<String, Row> takeCarInfoRdd = join.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, Tuple2<String, Row>>>, String, Row>() {
			private static final long serialVersionUID = -4900153459336468568L;

			@Override
			public Iterator<Tuple2<String, Row>> call(Iterator<Tuple2<String, Tuple2<String, Row>>> tuple2Iterator) throws Exception {
				List<Tuple2<String, Row>> list = new ArrayList<>();
				List<TakeCarAction> carActions = new ArrayList<>();
				SimpleDateFormat sdf = new SimpleDateFormat(ConfigEnum.MONITOR_ACTION_TIME_TYPE.getValue());
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, Tuple2<String, Row>> tuple2 = tuple2Iterator.next();
					list.add(new Tuple2<>(tuple2._1, tuple2._2._2));
					TakeCarAction tca = new TakeCarAction();
					tca.setTaskId(ConfigEnum.APPLICATION_TASK_ID.getLongValue());
					tca.setHour(tuple2._2._1.split(" ")[1]);
					tca.setDate(tuple2._2._2.getDate(0));
					tca.setMonitorId(tuple2._2._2.getString(1));
					tca.setCameraId(tuple2._2._2.getString(2));
					tca.setCar(tuple2._1);
					tca.setActionTime(sdf.parse(tuple2._2._2.getString(4)));
					tca.setSpeed(tuple2._2._2.getInt(5));
					tca.setRoadId(tuple2._2._2.getString(6));
					tca.setAreaId(tuple2._2._2.getString(7));
					carActions.add(tca);
				}
				saveList(carActions);
				return list.iterator();
			}
		});
		takeCarInfoRdd.count();

	}

	/**
	 * 计算每天各时间段要抽取的车的下标
	 *
	 * @param timeCarCount
	 * @return
	 */
	public static Map<String, Map<String, List<Integer>>> calculateTakeCarIndex(Map<String, Long> timeCarCount) {
		//date -> hour -> count
		Map<String, Map<String, Long>> dateTimeCount = new HashMap<>();
		timeCarCount.forEach((k, v) -> {
			String[] arr = k.split(" ");
			String date = arr[0];
			String hour = arr[1];
			Map<String, Long> tmp = dateTimeCount.get(date);
			if (tmp == null) {
				tmp = new HashMap<>();
				tmp.put(hour, v);
				dateTimeCount.put(date, tmp);
			} else {
				tmp.merge(hour, v, Long::sum);
			}
		});
		//Map<日期, Map<小时段, List<要抽取的车辆索引>>>
		Map<String, Map<String, List<Integer>>> carOffsetList = new HashMap<>();
		dateTimeCount.forEach((k, v) -> {
			long dayCount = v.values().stream().mapToLong(Long::longValue).sum();
			Map<String, List<Integer>> hourCarIndex = new HashMap<>();
			v.forEach((hour, count) -> {
				//按当天的比例从每个小时段中抽取，除法只有double和float才会保留小数，其他都是取整
				int hc = (int) ((double) count / dayCount * ConfigEnum.TAKE_SAMPLE_CAR_ONE_DAY.getIntValue());
				//如果抽取的数量大于当前时间段的总数量，就抽取全部
				if (hc >= count) {
					//取当前时间段的全部车辆
					hourCarIndex.put(hour, IntStream.range(0, count.intValue()).boxed().collect(Collectors.toList()));
				} else {
					List<Integer> carIndex = new ArrayList<>(hc);
					Random random = new Random();
					for (int i = 0; i < hc; ) {
						//从当前时间段随机取carIndex辆车
						int n = random.nextInt(count.intValue());
						if (!carIndex.contains(n)) {
							i++;
							carIndex.add(n);
						}
					}
					hourCarIndex.put(hour, carIndex);
				}
			});
			carOffsetList.put(k, hourCarIndex);
		});
		return carOffsetList;
	}

	public static void analysisMonitorTrafficPercentRate(SparkSession ss, JavaPairRDD<String, Row> monitorDetailRdd) {
		JavaPairRDD<String, Tuple2<String, Date>> carMonitorDateRdd = monitorDetailRdd.mapToPair(t -> new Tuple2<>(t._2.getString(3),
				new Tuple2(t._1, new SimpleDateFormat(ConfigEnum.MONITOR_ACTION_TIME_TYPE.getValue()).parse(t._2.getString(4)))));
		//对车辆记录分组聚合，key为car，value为车辆经过的monitor_id列表，已按照先后顺序排好序
		JavaPairRDD<String, List<String>> carMonitorsRdd = carMonitorDateRdd.groupByKey().mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, Iterable<Tuple2<String, Date>>>>, String, List<String>>() {
			private static final long serialVersionUID = -2384607495344417570L;

			@Override
			public Iterator<Tuple2<String, List<String>>> call(Iterator<Tuple2<String, Iterable<Tuple2<String, Date>>>> tuple2Iterator) throws Exception {
				List<Tuple2<String, List<String>>> list = new ArrayList<>();
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, Iterable<Tuple2<String, Date>>> next = tuple2Iterator.next();
					List<String> monitorIds = StreamSupport.stream(next._2.spliterator(), false).sorted(Comparator.comparing(Tuple2::_2))
							.map(Tuple2::_1).collect(Collectors.toList());
					list.add(new Tuple2<>(next._1, monitorIds));
				}
				return list.iterator();
			}
		});
		//0001,0002,0003,0004,0005，需要求：
		//0001,0002 ->0002 0001,0002
		//0001,0002,0003 ->0003 0001,0002,0003
		//0001,0002,0003,0004 ->0004 0001,0002,0003,0004
		//0001,0002,0003,0004,0005 ->0005 0001,0002,0003,0004,0005
		String[] configMonitors = ConfigEnum.MONITOR_CAR_CONVERSION_RATE.getValue().split(",");
		Map<String, List<String>> qryCond = new HashMap<>();
		for (int i = 0; i < configMonitors.length - 1; i++) {
			//subList返回的是不可序列化的RandomAccessSubList
			qryCond.put(configMonitors[i + 1], new ArrayList<>(Arrays.asList(configMonitors).subList(0, i + 2)));
		}
		log.info("monitor.car.conversion.rate.cond----" + qryCond);
		//广播需要统计的经过卡扣组合
		Broadcast<Map<String, List<String>>> broadcast = new JavaSparkContext(ss.sparkContext()).broadcast(qryCond);
		//累加器累加统计结果
		MonitorConversionAccumulator accumulator = new MonitorConversionAccumulator();
		ss.sparkContext().register(accumulator, "MonitorConversionAccumulator");
		carMonitorsRdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String, List<String>>>>() {
			private static final long serialVersionUID = -7957669057964020920L;

			@Override
			public void call(Iterator<Tuple2<String, List<String>>> tuple2Iterator) throws Exception {
				Map<String, List<String>> monitorMapList = broadcast.value();
				Map<String, Long> longMap = new HashMap<>();
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, List<String>> next = tuple2Iterator.next();
					monitorMapList.forEach((k, v) -> {
						if (next._2.contains(k)) {
							longMap.merge(k, 1L, Long::sum);
						}
						/*if (next._2.containsAll(v)) {
							longMap.merge(String.join(",", v), 1L, Long::sum);
						}*/
						//next._2： 0001->0002->0004->0003->0005
						//v:0001,0002,0003
						//判断车辆轨迹是否符合卡扣顺序，这里不要求车辆紧挨着经过卡扣，只需要确保经过的卡扣顺序不颠倒就行
						List<String> carTrack = next._2;
						int start = 0, count = 0;
						for (String s : v) {
							for (int j = start; j < carTrack.size(); j++) {
								if (s.equals(carTrack.get(j))) {
									start = j + 1;
									count++;
									break;
								}
							}
							if (start == 0) {
								break;
							}
						}
						if (count == v.size()) {
							longMap.merge(String.join(",", v), 1L, Long::sum);
						}
					});
				}
				accumulator.add(longMap);
			}
		});
		List<MonitorCarConversionRate> conversionRates = new ArrayList<>();
		Map<String, Long> monitorNums = accumulator.value();
		log.info("monitor.car.conversion.rate.cond monitorNums----" + monitorNums);
		//保存数据
		broadcast.value().forEach((k, v) -> {
			MonitorCarConversionRate conversionRate = new MonitorCarConversionRate(ConfigEnum.APPLICATION_TASK_ID.getLongValue(),
					ConfigEnum.MONITOR_CAR_CONVERSION_RATE.getValue());
			conversionRate.setFirstMonitorList(String.join(",", v));
			conversionRate.setFirstMonitorNum(monitorNums.get(conversionRate.getFirstMonitorList()));
			conversionRate.setSecondMonitorList(k);
			conversionRate.setSecondMonitorNum(monitorNums.get(k));
			conversionRate.setConversionRate();
			conversionRates.add(conversionRate);
		});
		saveList(conversionRates);
	}

	public static JavaPairRDD<Integer, String> checkMonitorState(SparkSession sparkSession, MonitorAccumulator monitorAccumulator, JavaPairRDD<String, MonitorCameraInfo> aggregateMonitorByIdRDD) {
		String sqlText = "select * from monitor_camera_info";
		JavaRDD<Row> baseRDD = sparkSession.sql(sqlText).toJavaRDD();
		JavaPairRDD<String, String> monitorId2CamerIdRDD = baseRDD.mapToPair(row -> new Tuple2<>(row.getString(0), row.getString(1)));
		//根据卡扣分组，样例数据：("0003", "cameraIds=02323,04563,03424|cameraCount=3")
		JavaPairRDD<String, MonitorCameraInfo> baseCameraInfoRDD = monitorId2CamerIdRDD.groupByKey().mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, String, MonitorCameraInfo>() {
			private static final long serialVersionUID = -4752785084624694674L;

			@Override
			public Tuple2<String, MonitorCameraInfo> call(Tuple2<String, Iterable<String>> values) throws Exception {
				// StringBuilder sb = new StringBuilder();
				// int count = 0;
				Set<String> cameraIds = new HashSet<>();
				for (String cameraId : values._2) {
					cameraIds.add(cameraId);
				}
				// String vs = "cameraIds=" + sb.substring(1) + "|" + "cameraCount=" + count;
				return new Tuple2<>(values._1, new MonitorCameraInfo(values._1, cameraIds, cameraIds.size()));
			}
		});
		//基本关系的RDD左连接抓怕记录的RDD
		JavaPairRDD<String, Tuple2<MonitorCameraInfo, Optional<MonitorCameraInfo>>> leftJoinRdd = baseCameraInfoRDD.leftOuterJoin(aggregateMonitorByIdRDD);
		JavaPairRDD<Integer, String> carCountMonitorRdd = leftJoinRdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, Tuple2<MonitorCameraInfo, Optional<MonitorCameraInfo>>>>, Integer, String>() {
			private static final long serialVersionUID = 268351518268824524L;

			@Override
			public Iterator<Tuple2<Integer, String>> call(Iterator<Tuple2<String, Tuple2<MonitorCameraInfo, Optional<MonitorCameraInfo>>>> tuple2Iterator) throws Exception {
				List<Tuple2<Integer, String>> list = new ArrayList<>();
				while (tuple2Iterator.hasNext()) {
					Tuple2<String, Tuple2<MonitorCameraInfo, Optional<MonitorCameraInfo>>> next = tuple2Iterator.next();
					String monitorId = next._1;
					//("0003", "cameraIds=02323,04563,03424|cameraCount=3")
					MonitorCameraInfo baseCameraInfo = next._2._1;
					// String monitorCameraInfoStr = "";
					Optional<MonitorCameraInfo> optional = next._2._2;
					MonitorCameraState mcs = new MonitorCameraState();
					if (optional.isPresent()) {
						//实际抓拍中有标准卡扣信息
						//("0005","monitorId=0005|camearIds=09200,03243|cameraCount=2|carCount==100")
						MonitorCameraInfo monitorCameraInfo = optional.get();
						// String[] split = monitorCameraInfoStr.split("\\|");
						//当前卡扣监测到的实际摄像头数量
						int factCameraCount = monitorCameraInfo.getCameraCount();
						if (factCameraCount == baseCameraInfo.getCameraCount()) {
							//当前卡扣监测到的摄像探头和基本表配置的数量一致
							//正常卡扣加1
							mcs.setNormalMonitorCount(1);
							//加上正常摄像头个数
							mcs.setNormalCameraCount(factCameraCount);
						} else {
							//当前卡扣监测到的摄像探头和基本表配置的数量不一致
							Set<String> factCamearIds = monitorCameraInfo.getCamearIds();
							Set<String> baseCamearIds = baseCameraInfo.getCamearIds();
							int baseCamearIdSize = baseCamearIds.size();
							baseCamearIds.removeAll(factCamearIds);
							mcs.setNormalCameraCount(baseCamearIdSize - baseCamearIds.size());
							mcs.setAbnormalMonitorCount(1);
							mcs.setAbnormalCameraCount(baseCamearIds.size());
							Map<String, Set<String>> info = new HashMap<>(2);
							info.put(monitorId, baseCamearIds);
							mcs.setAbnormalMonitorCameraInfo(info);
						}
						list.add(new Tuple2<>(monitorCameraInfo.getCarCount(), monitorCameraInfo.getMonitorId()));
					} else {
						//实际抓拍中没有标准卡扣信息
						// String[] split = baseCameraInfoStr.split("\\|");
						//abnormalMonitorCount=1|abnormalCameraCount=3|abnormalMonitorCameraInfos=0002:07553,07554
						// String countStr = "abnormalMonitorCount=1|abnormalCameraCount=" + split[1].split("=")[1] +
						// 		"|abnormalMonitorCameraInfos=" + monitorId + ":" + split[0];
						mcs.setAbnormalMonitorCount(1);
						mcs.setAbnormalCameraCount(baseCameraInfo.getCameraCount());
						Map<String, Set<String>> info = new HashMap<>(2);
						info.put(monitorId, baseCameraInfo.getCamearIds());
						mcs.setAbnormalMonitorCameraInfo(info);
					}
					monitorAccumulator.add(mcs);
				}
				//返回实际监测的数据，<车辆数，卡扣ID>
				return list.iterator();
			}
		});
		return carCountMonitorRdd;
	}

	public static void saveMonitorState(Long taskId, MonitorAccumulator monitorAccumulator) {
		MonitorCameraStateService service = ServiceFactory.buildService(MonitorCameraStateService.class);
		MonitorCameraState value = monitorAccumulator.value();
		value.setTaskId(taskId);
		service.insert(value);
		log.info("monitor result--------" + value);
		ServiceFactory.commitAndClose();
	}

	public static void saveMonitorState(List<MonitorFlowAction> monitorFlowActions) {
		MonitorCameraStateService service = ServiceFactory.buildService(MonitorCameraStateService.class);
		for (MonitorFlowAction m : monitorFlowActions) {
			service.insert(m);
		}
		ServiceFactory.commitAndClose();
	}

	public static void saveCarCount(List<MonitorCarCount> monitorCarCounts) {
		MonitorCameraStateService service = ServiceFactory.buildService(MonitorCameraStateService.class);
		for (MonitorCarCount m : monitorCarCounts) {
			service.insert(m);
		}
		ServiceFactory.commitAndClose();
	}

	public static void saveCarSpeedCount(List<MonitorCarSpeedCount> monitorCarSpeedCounts) {
		MonitorCameraStateService service = ServiceFactory.buildService(MonitorCameraStateService.class);
		for (MonitorCarSpeedCount m : monitorCarSpeedCounts) {
			service.insert(m);
		}
		ServiceFactory.commitAndClose();
	}

	public static <T> void saveList(List<T> list) {
		MonitorCameraStateService service = ServiceFactory.buildService(MonitorCameraStateService.class);
		for (T t : list) {
			service.insert(t);
		}
		ServiceFactory.commitAndClose();
	}

	static class TupleCompertor implements Comparator<Tuple2<String, Integer>>, Serializable {

		private static final long serialVersionUID = -7982194061521173575L;

		@Override
		public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
			return o2._2.compareTo(o1._2);
		}
	}
}
